package com.yzg.study.kafka.producer.controller;

import com.yzg.study.kafka.common.base.R;
import com.yzg.study.kafka.common.kafka.KafkaMsgUtils;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.Set;
import java.util.concurrent.ExecutionException;

@RestController
@Api(tags = "kafka消息发送者")
public class ProducerController {

    @Autowired
    private KafkaMsgUtils kafkaUtils;

    @PostMapping("/sendMsg")
    @ApiOperation("kafka发送消息给消费者")
    public R sendMsgToConsumer(@RequestParam("topic") String topic, @RequestParam("data") String data) {
        kafkaUtils.send(topic, data);
        return R.success();
    }

    @GetMapping("/topicList")
    @ApiOperation("获取kafka所有的主题列表")
    public R getTopicList(){
        try {
            Set<String> topicList = kafkaUtils.topicList();
            return R.success(topicList);
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
            return R.fail("获取kafka主题列表失败");
        }
    }
}
